package com.example.mq.mqserver.core;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
//通过这个类来实现消费者消费消息的逻辑
public class ConsumerManager {
    //持有virualhost来操作数据
    private VirtualHost parent;
    //指定一个线程池 执行具体的回调函数
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    //存放令牌的队列 实际上就是队列名 为了让线程池知道是哪一个队列的消息
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;
        scannerThread = new Thread(()->{
            while (true){
                try {
                    //1.拿到令牌
                    String queueName = tokenQueue.take();
                    //2.根据令牌 找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[ConsumerManager] 取令牌后发现 该队列名不存在 queueName = "+ queueName);
                    }
                    //3.从这个队列中消费一个消息
                    synchronized (queue){
                        consumerMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        //设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    //通知消费者消费消息
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null){
            throw new MqException("[ConsumerManager] 队列不存在 queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            //如果此时队列中已经有一些消息了 就需要立即消费
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                //这个方法调用一次就消费一条消息
                consumerMessage(queue);
            }
        }
    }

    private void consumerMessage(MSGQueue queue) {
        //消费消息
        //1.按照轮询的方式 找到消费者
        ConsumerEnv luckyDog = queue.chooseConsumerEnv();
        if(luckyDog == null){
            //当前对象没有消费者 暂时不需要 等有消费者再说
            return;
        }
        //从队列中取出一条消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            //当队列中还没有消息,也不需要消费
            return;
        }
        /**
         * 为了达成消息不丢失这样的效果
         * 1. 在真正执行回调之前 先把这个消息放在待确认集合中 避免因为回调失败 导致的消息丢失
         * 2. 真正执行回调
         * 3. 如消费者采用的是autoAck = true 默认回调函数执行结束之后不抛出异常 就算消费成功 然后就可以删除消息
         *    硬盘 内存消息中心的哈希表 上面的待确认消息集合
         * 4. 当前消费者采取的是autoAck = false 手动应答 需要消费者自己在自己的回调函数内部 调用basicAck这个API
         *
         */
        //把消息带入到消费者的回调方法中 去给线程池执行
        parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
        //真正执行回调
        workPool.submit(()->{
            try{
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                log.info("消息成功消费 queueName = " + queue.getName());
                //如果是自动应答 就可以直接把消息删除了
                if(luckyDog.isAutoAck()){
                    //删硬盘
                    if(message.getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //删待确认集合
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                    //删除内存中的消息中心里的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    log.info("消息成功消费 queueName :{}" ,queue.getName());
                }
                //如果是手动应答 先不处理 等消费者调用basicAck处理
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }
}
